package com.by.rpc.netty;

import com.by.log.Logger;
import com.by.log.LoggerFactory;
import com.by.log.TraceContext;
import com.by.rpc.dto.RpcRequest;
import com.by.rpc.dto.RpcResponse;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * Netty RPC客户端，用于发送RPC请求
 */
public class RpcClient {
    private static final Logger logger = LoggerFactory.getLogger(RpcClient.class);
    private static final int MAX_RETRY_TIMES = 3; // 最大重试次数
    private static final int RETRY_INTERVAL = 5; // 重试间隔秒数

    private final String host;
    private final int port;
    private volatile Channel channel;
    private final EventLoopGroup eventLoopGroup;
    // 存储请求ID与对应的CompletableFuture
    private final ConcurrentHashMap<String, CompletableFuture<RpcResponse>> requestMap;
    // 连接状态标志
    private volatile boolean isConnected = false;
    // 记录最后一次使用时间，供连接池判断空闲
    private volatile long lastUseTime;

    /**
     * 构造函数，仅初始化客户端实例，不执行阻塞连接
     */
    public RpcClient(String host, int port) {
        logger.info("Creating RpcClient for {}:{}", host, port);
        this.host = host;
        this.port = port;
        this.eventLoopGroup = new NioEventLoopGroup();
        this.requestMap = new ConcurrentHashMap<>();
        this.lastUseTime = System.currentTimeMillis();
    }

    /**
     * 异步连接到RPC服务端
     * 返回一个 CompletableFuture，当连接成功时完成
     */
    public CompletableFuture<Void> connectAsync() {
        CompletableFuture<Void> connectFuture = new CompletableFuture<>();
        logger.debug("Connecting to RPC server {}:{}", host, port);

        if (channel != null && channel.isActive()) {
            logger.debug("Channel is already active, skipping connection");
            connectFuture.complete(null);
            return connectFuture;
        }

        if (channel != null) {
            try {
                channel.close();
            } catch (Exception e) {
                logger.warn("Error closing existing channel: {}", e.getMessage());
            }
            channel = null;
        }

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                        pipeline.addLast(new LengthFieldPrepender(4));
                        pipeline.addLast(new KryoDecoder());
                        pipeline.addLast(new KryoEncoder());
                        pipeline.addLast(new RpcClientHandler(requestMap));
                    }
                });

        ChannelFuture future = bootstrap.connect(host, port);
        future.addListener((ChannelFutureListener) f -> {
            if (f.isSuccess()) {
                logger.info("Connected to RPC server: {}:{}", host, port);
                this.channel = f.channel();
                this.isConnected = true;
                connectFuture.complete(null);

                f.channel().closeFuture().addListener((ChannelFutureListener) future1 -> {
                    logger.info("RPC client disconnected from server {}:{}", host, port);
                    isConnected = false;
                });

            } else {
                logger.error("RPC client connection failed to {}:{}: {}", host, port, f.cause().getMessage(), f.cause());
                isConnected = false;
                connectFuture.completeExceptionally(f.cause());
            }
        });
        return connectFuture;
    }

    /**
     * 带重试机制的同步阻塞重连
     */
    public boolean retryConnect() {
        logger.info("Retrying connection to RPC server {}:{}", host, port);
        int retryTimes = 0;

        while (retryTimes < MAX_RETRY_TIMES) {
            try {
                logger.info("Retry connection attempt {} to {}:{}", ++retryTimes, host, port);
                connectAsync().get(5, TimeUnit.SECONDS);
                logger.info("Reconnected to RPC server successfully: {}:{}", host, port);
                return true;

            } catch (Exception e) {
                logger.error("Failed to reconnect to RPC server {}:{}", host, port, e);
                if (retryTimes < MAX_RETRY_TIMES) {
                    logger.info("Waiting {} seconds before next retry", RETRY_INTERVAL);
                    try {
                        TimeUnit.SECONDS.sleep(RETRY_INTERVAL);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }
        }
        logger.error("Failed to reconnect to RPC server after {} attempts: {}:{}", MAX_RETRY_TIMES, host, port);
        return false;
    }

    /**
     * 发送RPC请求
     */
    public CompletableFuture<RpcResponse> sendRequest(RpcRequest rpcRequest) {
        logger.debug("Sending RPC request: {}", rpcRequest);

        TraceContext traceContext = TraceContext.getCurrentContext();
        if (traceContext != null) {
            rpcRequest.setTraceId(traceContext.getTraceId());
            rpcRequest.setSpanId(traceContext.getSpanId());
            rpcRequest.setParentSpanId(traceContext.getParentSpanId());
        }

        if (channel == null || !channel.isActive() || !isConnected) {
            logger.warn("Channel is not active for {}:{}, reconnecting...", host, port);
            if (!retryConnect()) {
                logger.error("Failed to establish connection to {}:{}", host, port);
                CompletableFuture<RpcResponse> failedFuture = new CompletableFuture<>();
                failedFuture.completeExceptionally(new RuntimeException(
                        "无法连接到RPC服务: " + host + ":" + port +
                                "，请检查服务是否正常运行或网络连接是否正常"));
                return failedFuture;
            }
        }

        if (channel == null || !channel.isActive()) {
            logger.error("Channel is null or inactive, cannot send request to {}:{}", host, port);
            CompletableFuture<RpcResponse> failedFuture = new CompletableFuture<>();
            failedFuture.completeExceptionally(new RuntimeException(
                    "RPC服务连接不可用: " + host + ":" + port +
                            "，通道未激活或为空"));
            return failedFuture;
        }

        CompletableFuture<RpcResponse> responseFuture = new CompletableFuture<>();
        requestMap.put(rpcRequest.getRequestId(), responseFuture);

        channel.writeAndFlush(rpcRequest).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                logger.debug("RPC request sent successfully: {}", rpcRequest.getRequestId());
                this.lastUseTime = System.currentTimeMillis();
            } else {
                String errorMsg = "发送RPC请求失败: " + future.cause().getMessage() +
                        ", 请求ID: " + rpcRequest.getRequestId() +
                        ", 服务地址: " + host + ":" + port;
                logger.error(errorMsg, future.cause());
                requestMap.remove(rpcRequest.getRequestId());
                responseFuture.completeExceptionally(new RuntimeException(errorMsg, future.cause()));
            }
        });

        return responseFuture;
    }

    /**
     * 关闭客户端
     */
    public void close() {
        logger.info("Closing RpcClient for {}:{}", host, port);
        isConnected = false;
        if (channel != null) {
            try {
                channel.close().sync();
            } catch (InterruptedException e) {
                logger.warn("Interrupted while closing channel", e);
                Thread.currentThread().interrupt();
            }
        }
        eventLoopGroup.shutdownGracefully();
    }

    /**
     * 检查连接是否活跃
     */
    public boolean isActive() {
        boolean active = isConnected && channel != null && channel.isActive();
        logger.debug("Client {}:{} is active: {}", host, port, active);
        if (isConnected && (channel == null || !channel.isActive())) {
            isConnected = false;
            logger.debug("Updated connection status for {}:{} to inactive", host, port);
        }
        return active;
    }

    /**
     * 获取最后一次使用时间
     */
    public long getLastUseTime() {
        return lastUseTime;
    }
}